package com.zg.CliemtManage;

import com.alibaba.fastjson2.JSON;
import com.zg.domain.DataForClient;
import com.zg.domain.LotClient;
import com.zg.utils.ServerPublishClient;
import io.netty.channel.Channel;
import io.netty.handler.codec.mqtt.MqttPublishMessage;
import io.netty.handler.codec.mqtt.MqttQoS;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 主题管理
 */
@Slf4j
@Component
public class ClientTopics {


    private ServerPublishClient serverPublishClient = new ServerPublishClient();

    /**
     * 采取的是topic----------》对应客户端列表
     */
    private ConcurrentHashMap<String, LinkedList<Channel>> topic_client = DataForClient.getTopic_Client();

    /**
     * 统一管理订阅列表
     * @param topics
     * @param channel
     */
    public void addTopic(Set<String> topics, Channel channel){
        for (String topic : topics) {
            if (!topic_client.containsKey(topic)){
                LinkedList<Channel> list = new LinkedList<>();
                list.add(channel);
                topic_client.put(topic,list);
                log.info("添加新主题：{}",topic);
            }else {
                LinkedList<Channel> list = topic_client.get(topic);
                list.add(channel);
                log.info("向主题:{},添加客户端通道，一共有：{}个订阅客户端",topic,list.size());
            }
        }
    }


    /**
     * 主题转发消息
     * @param topic
     * @param data
     */
    public void sendTopicMsg(String topic,Object data){
        Object send;
        if (data instanceof String){
            HashMap hashMap = JSON.parseObject((String) data, HashMap.class);
            send = hashMap;
        }else {
            send = data;
        }
        topic_client.forEach((key,value)->{
             if (startWith(key, topic)){
                 log.info("转发到的主题：{}",key);
                 for (Channel channel : value) {
                     if (channel.isActive()){
                         Random random = new Random();
                         int i = random.nextInt(999) + 1;
                         MqttPublishMessage mqttPublishMessage = serverPublishClient.pushMsgToClient(topic, send, i, false, MqttQoS.AT_MOST_ONCE, false);
                         log.info("发送消息给channel：{}",channel);
                         channel.writeAndFlush(mqttPublishMessage);
                     }else {
                         log.error("客户端：{}发生异常不在线",channel);
                     }

                 }
             }
        });
        /*if (topic_client.containsKey(topic)){
            LinkedList<Channel> list = topic_client.get(topic);
            for (Channel channel : list) {
                if (channel.isActive()){
                    Random random = new Random();
                    int i = random.nextInt(999) + 1;
                    MqttPublishMessage mqttPublishMessage = serverPublishClient.pushMsgToClient(topic, hashMap, i, false, MqttQoS.AT_MOST_ONCE, false);
                    log.info("发送消息给channel：{}",channel);
                    channel.writeAndFlush(mqttPublishMessage);
                }else {
                    log.error("客户端：{}发生异常不在线",channel);
                }

            }
        }*/
    }

    /**
     * 移除设备相关主题
     * 使用在设备下线的时候
     * @param lotClient
     */
    public void delClientAllTopic(LotClient lotClient){
        List<String> topics = lotClient.getTopics();
        Channel channel = lotClient.getChannel();
        for (String topic : topics) {
            if (topic_client.containsKey(topic)){
                LinkedList<Channel> list = topic_client.get(topic);
                list.remove(channel);//移除相关主题
                log.info("将客户端id：{}的：{}主题移除,订阅量还有:{}",lotClient.getClient_id(),topic,list.size());
            }
        }
    }

    /**
     * 取消订阅一些主题
     * 用于客户端主动取消一些主题以及管理设备取消某个设备的主题
     * @param channel
     * @param topic
     */
    public void delSomeTopic(Channel channel, List<String> topic){
        for (String s : topic) {
            if (topic_client.containsKey(s)){
                LinkedList<Channel> list = topic_client.get(s);
                list.remove(channel);
                log.info("channel:{}取消主题：{}",channel,s);
            }
        }
    }


    /**
     * 获取符合条件的主题
     * @param topic
     * @param topics//服务端存储的topic
     * @return
     */
    public boolean startWith(String topic, String topics){
        String[] split = topic.split("/");
        StringBuilder a = new StringBuilder();//保存主题开始路径
        for (String s : split) {
            a.append("/");
            if (s.equals("#")){
                break;
            }else {
                a.append(s);
            }
        }
        a.deleteCharAt(0);
        String start = a.toString();
        if (topics.startsWith(start)){
            return true;
        }
        return false;
    }

    /**
     * 发送到设备的指定主题上去
     * @param topic
     * @param data
     * @param channel
     */
    public void sendMsgToClient(String topic ,Object data,Channel channel) {
        Random random = new Random();
        int i = random.nextInt(999) + 1;
        MqttPublishMessage mqttPublishMessage = serverPublishClient.pushMsgToClient(topic, data, i, false, MqttQoS.AT_MOST_ONCE, false);
        channel.writeAndFlush(mqttPublishMessage);
    }
}
